package com.lvmama.rhino.analyze.processing

import com.lvmama.rhino.common.entity._
import com.lvmama.rhino.common.utils.JDBCUtil.Conversions
import com.lvmama.rhino.common.utils.Utils
import org.apache.spark.SparkContext
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions.{col, _}


/**
  * 坑位产品订单
  *
  */
object WirelessPitClick {

  import Conversions._

  def process(sc: SparkContext, pageForward: DataFrame): Unit = {

    val commons = Seq("logType", "sessionId", "deviceToken", "channelCode", "platformCode", "pageTypeCode", "timestamp", "pageParam", "holeCode", "accessCode", "buttonCode", "ip")
    val yesterday = Utils.getYesterday()
    //专题坑位点击
    val resultDataFrame = pageForward.filter(col("pageTypeCode").contains(TOPIC.code))
      .select(commons.map(col): _*)
      .withColumn("area", col("pageParam").getItem("area"))
      .withColumn("navName", col("pageParam").getItem("navName"))
      .withColumn("tabName", col("pageParam").getItem("tabName"))
      .withColumn("oi", col("pageParam").getItem("oi"))
      .coalesce(20)

    //专题页面PV
    resultDataFrame.filter(col("logType").equalTo(LogType.forward.toString))
      .filter(col("area").equalTo("") && col("navName").equalTo("") )
      .groupBy(col("platformCode"),col("pageTypeCode")).agg(count("*").as("click_num"))
      .select(col("platformCode").as("platform_code"),col("pageTypeCode").as("code"),col("click_num"),
      lit(yesterday).as("oper_date"))
    .insertDF2MysqlDirect("flow_stat_topic_detail")

    //nav曝光量
    val navSlide = resultDataFrame.filter(col("navName") !== "")
      .filter(col("pageTypeCode").contains(TOPIC.code))
      .filter(col("buttonCode").equalTo(BottonCode.slide.toString))
      .groupBy(col("platformCode"),col("pageTypeCode"), col("area"), col("navName")).agg(count("*").as("navSlide"))
      .select(col("platformCode").as("platform_code"),col("pageTypeCode").as("code"), col("area").as("nav_area"),
        col("navName").as("nav_name"), col("navSlide").as("nav_slide"))

    //nav点击量
    val navPV = resultDataFrame.filter(col("navName") !== "")
      .filter(col("pageTypeCode").contains(TOPIC.code))
      .filter(col("oi") !== "")
      .groupBy(col("platformCode"),col("pageTypeCode"), col("area"), col("navName"), col("tabName"))
      .agg(count("*").as("nav_click"))

    val navUV = resultDataFrame.filter(col("navName") !== "")
      .filter(col("pageTypeCode").contains(TOPIC.code))
      .filter(col("oi") !== "")
      .dropDuplicates(Seq("ip","platformCode", "pageTypeCode", "area", "navName", "tabName"))
      .groupBy(col("platformCode"),col("pageTypeCode"), col("area"), col("navName"), col("tabName"))
      .agg(count("*").as("user_click"))

    val navFlow = navUV.join(navPV, Seq("platformCode","pageTypeCode", "area", "navName", "tabName"))
      .select(navUV.col("platformCode"),navUV.col("pageTypeCode"), navUV.col("area"), navUV.col("navName"), navUV.col("tabName"), col("nav_click"), col("user_click"))

    val nav = navSlide.join(navFlow, navSlide("platform_code") === navFlow("platformCode") && navSlide("code") === navFlow("pageTypeCode") && navSlide("nav_area") === navFlow("area")
      && navSlide("nav_name") === navFlow("navName"), "left_outer")
      .select(col("platform_code").as("platform_code"),col("code").as("nav_code"),
        when(col("nav_area").isNotNull, col("nav_area")).otherwise("").as("nav_area"),
        col("nav_name"),
        when(col("tabName").isNotNull, col("tabName")).otherwise("").as("tab_name"),
        col("nav_slide"),
        when(col("nav_click").isNotNull, col("nav_click")).otherwise("0").as("nav_click"),
        col("user_click"))

    //产品点击
    var product = resultDataFrame.filter(col("pageTypeCode").contains(TOPIC.code))
      .filter(!col("navName").equalTo(""))
      .select(col("platformCode"),col("pageTypeCode").as("code"), col("area"), col("navName"), col("tabName"), col("timestamp").as("time"),
        when(col("platformCode").equalTo("2g53"), col("sessionId")).otherwise(col("deviceToken")).as("session_id"))

    //产品待支付
    var order = pageForward.filter(col("pageTypeCode").equalTo(PENDING_PAYMENT.code) && col("accessCode").contains(TOPIC.code))
      .select(commons.map(col): _*)
      .withColumn("orderNum", col("pageParam").getItem("orderNum"))
      .withColumn("paymentAmount", col("pageParam").getItem("paymentAmount"))
      .orderBy(col("timestamp").desc)
      .select(col("platformCode"),col("accessCode"), col("timestamp"),
        when(col("platformCode").equalTo("2g53"), col("sessionId")).otherwise(col("deviceToken")).as("sessionId"),
        col("orderNum"), col("paymentAmount").as("order_amount"))
      .dropDuplicates("orderNum")

    //产品支付
    var pay = pageForward.filter(col("pageTypeCode").equalTo(COMPLETED_PAYMENT.code) && col("accessCode").contains(TOPIC.code))
      .withColumn("orderNum", col("pageParam").getItem("orderNum"))
      .withColumn("paymentAmount", col("pageParam").getItem("paymentAmount"))
      .select(col("orderNum").as("payNum"), col("paymentAmount").as("pay_amount"))
      .dropDuplicates("payNum")

    var revenue = order.join(pay, order("orderNum") === pay("payNum"), "left_outer")
      .select(col("platformCode").as("platform_code"),col("accessCode"), col("sessionId"), col("timestamp"), col("order_amount"), col("pay_amount"))

    //产品点击与订单关联
    var POT = revenue.join(product, product("code") === revenue("accessCode") && product("session_id") === revenue("sessionId"), "left_outer")
      .filter(col("time") < col("timestamp"))
      .groupBy(col("platform_code"),col("code").as("CODE"), col("sessionId").as("SID"))
      .agg(max("time").as("time1"), max("timestamp").as("time2"))

    var productOrder = POT.join(product, product("platformCode") === POT("platform_code") && product("code") === POT("CODE") && product("session_id") === POT("SID") && product("time") === POT("time1"))
      .join(revenue,product("platformCode") === POT("platform_code") && revenue("accessCode") === POT("CODE") && revenue("sessionId") === POT("SID") && revenue("timestamp") === POT("time2"))
      .select(POT.col("platform_code"),POT.col("CODE").as("CODE"), col("area"), col("navName"), col("tabName"), col("order_amount"), col("pay_amount"))

    nav.join(productOrder, nav("platform_code") === productOrder("platform_code") && nav("nav_code") === productOrder("CODE")
      && nav("nav_area") === productOrder("area") && nav("nav_name") === productOrder("navName")
      && nav("tab_name") === productOrder("tabName"), "left_outer")
      .select(nav.col("platform_code"),col("nav_code").as("code"), col("nav_area").as("area"),
        col("nav_name"), col("tab_name"), col("nav_slide").as("slide_num"), col("nav_click").as("click_num"), col("user_click"),
        when(col("order_amount").isNotNull, col("order_amount")).as("order_money"),
        when(col("pay_amount").isNotNull, col("pay_amount")).as("pay_money"),
        lit(yesterday).as("oper_date"))
      .insertDF2MysqlDirect("flow_stat_topic_detail")

    //坑位点击量
    val pit = pageForward.filter(col("logType").equalTo(LogType.click.toString))
      .filter(col("pageTypeCode").contains(CHANNEL.code))
      .filter(!col("holeCode").equalTo("kwnull"))
      .select(commons.map(col): _*)

    val pitPV = pit.groupBy(col("platformCode"), col("channelCode"), col("holeCode")).agg(count("*").as("pv"))

    val pitUV = pit.dropDuplicates(Seq("platformCode", "channelCode", "holeCode", "ip"))
      .groupBy(col("platformCode"), col("channelCode"), col("holeCode")).agg(count("*").as("uv"))

    pitUV.join(pitPV, Seq("platformCode", "channelCode", "holeCode"))
      .select(pitUV.col("platformCode").as("platform_code"), pitUV.col("channelCode").as("channel_code"),
        pitUV.col("holeCode").as("code"), col("pv").as("click_num"), col("uv").as("user_click"),
        lit(yesterday).as("oper_date"))
      .insertDF2MysqlDirect("flow_stat_pit_click")

    pageForward.filter(col("logType").equalTo(LogType.forward.toString))
      .filter(col("pageTypeCode").contains("3PinDuau"))
      .groupBy(col("platformCode"), col("channelCode"), col("pageTypeCode")).agg(count("*").as("click_num"))
      .select(col("platformCode").as("platform_code"), col("channelCode").as("channel_code"), col("pageTypeCode").as("code"), col("click_num"),
        lit(yesterday).as("oper_date"))
      .insertDF2MysqlDirect("flow_stat_pit_click")

  }

}
